package io.task.cc.message.socket.spi.client;

import io.task.cc.message.socket.*;
import io.task.cc.message.socket.lock.SetWithLock;
import io.task.cc.message.socket.spi.client.intf.ClientHandler;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.StandardSocketOptions;
import java.nio.channels.AsynchronousChannelGroup;
import java.nio.channels.AsynchronousSocketChannel;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;


@Slf4j
@Setter
@Getter
public class Client {

    private final AsynchronousChannelGroup channelGroup;

    private ClientConfig tioClientConfig;


    public Client(final ClientConfig tioClientConfig) throws IOException {
        super();
        this.tioClientConfig = tioClientConfig;
        this.channelGroup = AsynchronousChannelGroup.withThreadPool(tioClientConfig.groupExecutor);

        startHeartbeatTask();
        startReconnTask();
    }


    public void asynConnect(Node serverNode) throws Exception {
        asynConnect(serverNode, null);
    }


    public void asynConnect(Node serverNode, Integer timeout) throws Exception {
        asynConnect(serverNode, null, null, timeout);
    }


    public void asynConnect(Node serverNode, String bindIp, Integer bindPort, Integer timeout) throws Exception {
        connect(serverNode, bindIp, bindPort, null, timeout, false);
    }


    public ClientChannelContext connect(Node serverNode) throws Exception {
        return connect(serverNode, null);
    }


    public ClientChannelContext connect(Node serverNode, Integer timeout) throws Exception {
        return connect(serverNode, null, 0, timeout);
    }

    /**
     * @param serverNode
     * @param bindIp
     * @param bindPort
     * @param initClientChannelContext
     * @param timeout                  超时时间，单位秒
     * @return
     * @throws Exception
     * @author tanyaowu
     */
    public ClientChannelContext connect(Node serverNode, String bindIp, Integer bindPort, ClientChannelContext initClientChannelContext, Integer timeout) throws Exception {
        return connect(serverNode, bindIp, bindPort, initClientChannelContext, timeout, true);
    }

    /**
     * @param serverNode
     * @param bindIp
     * @param bindPort
     * @param initClientChannelContext
     * @param timeout                  超时时间，单位秒
     * @param isSyn                    true: 同步, false: 异步
     * @return
     * @throws Exception
     * @author tanyaowu
     */
    private ClientChannelContext connect(Node serverNode, String bindIp, Integer bindPort, ClientChannelContext initClientChannelContext, Integer timeout, boolean isSyn)
            throws Exception {

        AsynchronousSocketChannel asynchronousSocketChannel = null;
        ClientChannelContext channelContext = null;
        boolean isReconnect = initClientChannelContext != null;
        //		TioClientListener tioClientListener = tioClientConfig.getTioClientListener();

        long start = System.currentTimeMillis();
        asynchronousSocketChannel = AsynchronousSocketChannel.open(channelGroup);
        long end = System.currentTimeMillis();
        long iv = end - start;
        if (iv >= 100) {
            log.error("{}, open 耗时:{} ms", channelContext, iv);
        }

        asynchronousSocketChannel.setOption(StandardSocketOptions.TCP_NODELAY, true);
        asynchronousSocketChannel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
        asynchronousSocketChannel.setOption(StandardSocketOptions.SO_KEEPALIVE, true);

        InetSocketAddress bind = null;
        if (bindPort != null && bindPort > 0) {
            if (bindIp != null) {
                bind = new InetSocketAddress(bindIp, bindPort);
            } else {
                bind = new InetSocketAddress(bindPort);
            }
        }

        if (bind != null) {
            asynchronousSocketChannel.bind(bind);
        }

        channelContext = initClientChannelContext;

        start = System.currentTimeMillis();

        InetSocketAddress inetSocketAddress = new InetSocketAddress(serverNode.getIp(), serverNode.getPort());

        ConnectionCompletionVo attachment = new ConnectionCompletionVo(channelContext, this, isReconnect, asynchronousSocketChannel, serverNode, bindIp, bindPort);

        if (isSyn) {
            Integer realTimeout = timeout;
            if (realTimeout == null) {
                realTimeout = 5;
            }

            CountDownLatch countDownLatch = new CountDownLatch(1);
            attachment.setCountDownLatch(countDownLatch);

            try {
                asynchronousSocketChannel.connect(inetSocketAddress, attachment, tioClientConfig.getConnectionCompletionHandler());
            } catch (Throwable e) {
                tioClientConfig.getConnectionCompletionHandler().failed(e, attachment);
                return attachment.getChannelContext();
            }

            @SuppressWarnings("unused")
            boolean f = countDownLatch.await(realTimeout, TimeUnit.SECONDS);
            return attachment.getChannelContext();
        } else {
            try {
                asynchronousSocketChannel.connect(inetSocketAddress, attachment, tioClientConfig.getConnectionCompletionHandler());
            } catch (Throwable e) {
                tioClientConfig.getConnectionCompletionHandler().failed(e, attachment);
            }
            return null;
        }
    }

    /**
     * @param serverNode
     * @param bindIp
     * @param bindPort
     * @param timeout    超时时间，单位秒
     * @return
     * @throws Exception
     * @author tanyaowu
     */
    public ClientChannelContext connect(Node serverNode, String bindIp, Integer bindPort, Integer timeout) throws Exception {
        return connect(serverNode, bindIp, bindPort, null, timeout);
    }


    /**
     * @param channelContext
     * @param timeout        单位秒
     */
    public void reconnect(ClientChannelContext channelContext, Integer timeout) throws Exception {
        connect(channelContext.getServerNode(), channelContext.getBindIp(), channelContext.getBindPort(), channelContext, timeout);
    }


    /**
     * 定时任务：发心跳
     */
    private void startHeartbeatTask() {

        final ClientHandler tioHandler = tioClientConfig.getTioClientHandler();

        final String id = tioClientConfig.getId();
        new Thread(new Runnable() {
            @Override
            public void run() {
                while (!tioClientConfig.isStopped()) {
                    //					final long heartbeatTimeout = tioClientConfig.heartbeatTimeout;
                    if (tioClientConfig.heartbeatTimeout <= 0) {
                        log.warn("用户取消了框架层面的心跳定时发送功能，请用户自己去完成心跳机制");
                        break;
                    }
                    SetWithLock<ChannelContext> setWithLock = tioClientConfig.connecteds;
                    ReadLock readLock = setWithLock.readLock();
                    readLock.lock();
                    try {
                        Set<ChannelContext> set = setWithLock.getObj();
                        long currtime = System.currentTimeMillis();
                        for (ChannelContext entry : set) {
                            ClientChannelContext channelContext = (ClientChannelContext) entry;
                            if (channelContext.isClosed || channelContext.isRemoved) {
                                continue;
                            }

                            ChannelState stat = channelContext.stat;
                            long compareTime = Math.max(stat.latestTimeOfReceivedByte, stat.latestTimeOfSentPacket);
                            long interval = currtime - compareTime;
                            if (interval >= tioClientConfig.heartbeatTimeout / 2) {
                                Packet packet = tioHandler.heartbeatPacket(channelContext);
                                if (packet != null) {
                                    if (log.isInfoEnabled()) {
                                        log.info("{}发送心跳包", channelContext);
                                    }
                                    Io.send(channelContext, packet);
                                }
                            }
                        }


                    } catch (Throwable e) {
                        log.error("", e);
                    } finally {
                        try {
                            readLock.unlock();
                            Thread.sleep(tioClientConfig.heartbeatTimeout / 4);
                        } catch (Throwable e) {
                            log.error(e.toString(), e);
                        } finally {

                        }
                    }
                }
            }
        }, "tio-timer-heartbeat" + id).start();
    }

    /**
     * 启动重连任务
     */
    private void startReconnTask() {
        final ReconnConf reconnConf = tioClientConfig.getReconnConf();
        if (reconnConf == null || reconnConf.getInterval() <= 0) {
            return;
        }

        final String id = tioClientConfig.getId();
        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                while (!tioClientConfig.isStopped()) {
                    log.error("closeds:{}, connections:{}", tioClientConfig.closeds.size(), tioClientConfig.connections.size());
                    //log.info("准备重连");
                    LinkedBlockingQueue<ChannelContext> queue = reconnConf.getQueue();
                    ClientChannelContext channelContext = null;
                    try {
                        channelContext = (ClientChannelContext) queue.take();
                    } catch (InterruptedException e1) {
                        log.error(e1.toString(), e1);
                    }
                    if (channelContext == null) {
                        continue;
                        //						return;
                    }

                    if (channelContext.isRemoved) //已经删除的，不需要重新再连
                    {
                        continue;
                    }


                    long sleeptime = reconnConf.getInterval() - (System.currentTimeMillis() - channelContext.stat.timeInReconnQueue);
                    //log.info("sleeptime:{}, closetime:{}", sleeptime, timeInReconnQueue);
                    if (sleeptime > 0) {
                        try {
                            Thread.sleep(sleeptime);
                        } catch (InterruptedException e) {
                            log.error(e.toString(), e);
                        }
                    }

                    if (channelContext.isRemoved || !channelContext.isClosed) //已经删除的和已经连上的，不需要重新再连
                    {
                        continue;
                    } else {
                        ReconnRunnable runnable = channelContext.getReconnRunnable();
                        if (runnable == null) {
                            synchronized (channelContext) {
                                runnable = channelContext.getReconnRunnable();
                                if (runnable == null) {
                                    runnable = new ReconnRunnable(channelContext, Client.this, reconnConf.getThreadPoolExecutor());
                                    channelContext.setReconnRunnable(runnable);
                                }
                            }
                        }
                        runnable.execute();
                        //						reconnConf.getThreadPoolExecutor().execute(runnable);
                    }
                }
            }
        });
        thread.setName("tio-timer-reconnect-" + id);
        thread.setDaemon(true);
        thread.start();

    }


    public boolean stop() {
        boolean ret = true;
        try {
            tioClientConfig.groupExecutor.shutdown();
        } catch (Exception e1) {
            log.error(e1.toString(), e1);
        }
        try {
            tioClientConfig.tioExecutor.shutdown();
        } catch (Exception e1) {
            log.error(e1.toString(), e1);
        }

        tioClientConfig.setStopped(true);
        try {
            ret = ret && tioClientConfig.groupExecutor.awaitTermination(6000, TimeUnit.SECONDS);
            ret = ret && tioClientConfig.tioExecutor.awaitTermination(6000, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            log.error(e.getLocalizedMessage(), e);
        }
        log.info("client resource has released");
        return ret;
    }
}
